package com.xrui.hbase.tools;


import com.xrui.hbase.*;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;


/**
 * Created by U6034830 on 6/20/2017.
 */
public class Test {
    private static final Logger logger = LoggerFactory.getLogger(Test.class);
    private static final String KERBOROS_PRINCIPAL = "bigdata-app-catoftiger-hbasethrift@INTQA.THOMSONREUTERS.COM";
    private static final String KERBOROS_KEYTAB = "E:/Jobs/hbaseapi/hbase_tool_prototype_0.1/hbase/conf/kerboros-keytab.keytab";

    public static void test() {
        String table = "catoftiger:100org_asrbal_global_0.0.1";

        HBase hbase = null;
        HBaseTable hTable = null;
        TableReader hReader = null;
        try {
            System.setProperty("hadoop.home.dir", "E:/Jobs/hbaseapi/hbase_tool_prototype_0.1/hbase/");
            System.setProperty("java.security.krb5.conf", "E:/Jobs/hbaseapi/hbase_tool_prototype_0.1/hbase/conf/krb5.ini");
            //System.setProperty("java.security.auth.login.config","D:/Program Files/hbase-1.1.10/conf/jaas.conf");
            // Create connection to HBase
            logger.info("Establishing connection to HBase...");

            TableName tableName = TableName.valueOf(table);
            String uri = "hbase://.env/default/catoftiger:100org_asrbal_global_0.0.1/cf?" +
                "FundamentalSeriesId=48433149979&" +
                //"PeriodEndDate=1993-12-31T00:00:00Z&" +
                //"PeriodType=Year&" +
                //"CumulativeType=N&" +
                //"StatementTypeCode=BAL&" +
                //"FinancialConceptGlobal=AACR" +
                "#exporttocsv";
            HBaseURI hbaseURI = HBaseURI.ofValue(uri);

            hbase = HBase.open(hbaseURI);

            UserGroupInformation.setConfiguration(hbase.getConfiguration());
            UserGroupInformation gui = UserGroupInformation.loginUserFromKeytabAndReturnUGI(KERBOROS_PRINCIPAL, KERBOROS_KEYTAB);
            UserGroupInformation.setLoginUser(gui);

            hTable = hbase.openTable(table.toString());
            hReader = hTable.openTableReader();

            //case1. get

            EntityId entityId = hTable.getEntityId("48433149979",
                "1993-12-31T00:00:00Z",
                "Year",
                "N",
                "BAL",
                "AACR");
            DataRequest dataRequest = DataRequest.create("cf");
            RowData rowData = hReader.get(entityId, dataRequest);

            //case2. batch get
            /*EntityId entityId2_1 = hTable.getEntityId("48433149979",
                "1993-12-31T00:00:00Z",
                "Year",
                "N",
                "BAL",
                "AACR");
            EntityId entityId2_2 = hTable.getEntityId("48433149979",
                "1993-12-31T00:00:00Z",
                "Year",
                "N",
                "BAL",
                "AACR");
            DataRequest dataRequest2 = DataRequest.create("cf");
            java.util.List<EntityId> entityIdList2 = new ArrayList<EntityId>(2);;

            entityIdList2.add(entityId2_1);
            entityIdList2.add(entityId2_2);

            java.util.List<RowData> rowDataList2 = hReader.bulkGet(entityIdList2, dataRequest2);
            for (RowData rowData2: rowDataList2) {
                logger.info(rowData2.getEntityId().toString());
            }*/

            //case3. scan - with no qulifier
            /*
            ColumnName columnName3 = ColumnName.create("cf");
            DataRequestBuilder.ColumnsDef columnsDef3 = DataRequestBuilder.ColumnsDef.create().withMaxVersions(2)
                                                            .withPageSize(10).add(columnName3);
            DataRequest dataRequest3 = DataRequest.builder().addColumns(columnsDef3).build();

            RowScanner rowScanner3 = hReader.getScanner(dataRequest3);
            for (RowData rowData3: rowScanner3) {
                logger.info(rowData3.getEntityId().toString());
            }
            */

            //case4. scan

            ColumnName columnName4 = ColumnName.create("cf");
            DataRequestBuilder.ColumnsDef columnsDef4 = DataRequestBuilder.ColumnsDef.create().withMaxVersions(2)
                .withPageSize(10).add(columnName4);
            DataRequest dataRequest4 = DataRequest.builder().addColumns(columnsDef4).build();

            EntityId startEntityId = hTable.getEntityId("48433149979",
                "1993-12-31T00:00:00Z",
                "Year",
                "N",
                "BAL",
                "AACR");

            EntityId endEntityId = hTable.getEntityId("48433149979",
                "1993-12-31T00:00:00Z",
                "Year",
                "N",
                "BAL",
                "AACR");

            TableReader.ScannerOptions scannerOptions4 = new TableReader.ScannerOptions().setStartRow(startEntityId)
                .setStopRow(endEntityId)
                .setRowCaching(10);
                //.setRowFilter();
            RowScanner rowScanner4 = hReader.getScanner(dataRequest4, scannerOptions4);
            for (RowData rowData4: rowScanner4) {
                logger.info(rowData4.getEntityId().toString());
            }


        } catch (Exception e) {
            logger.error("Unknown error occurred", e);
        } finally {
            if (hReader != null) {
                try {
                    hReader.close();
                } catch (Exception e) {
                    logger.warn("Close Reader Failed with: ", e);
                }
            }
            if (hbase != null) {
                try {
                    hbase.release();
                } catch (IOException e) {
                    logger.warn("Error occurred when release connection to HBase: ", e);
                }
            }
            if (hTable != null) {
                try {
                    hTable.release();
                } catch (IOException e) {
                    logger.warn("Error occurred when release connection to HBase table: ", e);
                }
            }
        }

    }

    public static void run(String[] args) {
        logger.info("--run the tests--");
        logger.info("--end the tests--");
        try {
            String Quorum[] = new String[1];
            Quorum[0] = "10.35.20.149";
            HBaseURI hbaseURI = HBaseURI.newBuilder()
                .withZookeeperQuorum(Quorum)
                .withZookeeperClientPort(2181)
                .withTableName("catoftiger:100org_asrbal_global")
                .addColumnName(ColumnName.create("cf:qualifier"))
                .build();

            HBase hbase = HBaseFactory.get().open(hbaseURI);
            //TablePool pool = TablePool.newBuilder(hbase)
            //    .withMaxSize(2)
            //.withMinSize(0)
            //.withIdlePollPeriod(0)
            //.withIdleTimeout(0)
            //s   .build();
            //Table table = pool.get("Spark_RuleChange");
            Table table = hbase.openTable("Spark_RuleChange");
            TableReader tableReader = table.openTableReader();
            RowScanner scan = tableReader.getScanner(DataRequest.builder().build());
            for (RowData row : scan) {
                row.getEntityId();
                row.getCells("family", "qualifier").get("key").getData();
            }
        } catch (Exception e) {
            logger.error(e.toString());
        } finally {

        }







/*
        ""final KijiURI kijiURI = KijiURI.newBuilder().withInstanceName("kiji_instance_name").build();""
        ""final Kiji kiji = Kiji.Factory.open(kijiURI);""
        ""final KijiTablePool pool = KijiTablePool.newBuilder(kiji)""
        ""    .withIdleTimeout(10)""
        ""    .withIdlePollPeriod(1)""
        ""    .build();""
        """"
        ""KijiTable fooTable1 = pool.get("foo");""
        ""KijiTable fooTable2 = pool.get("foo");// there will be 2 connections for table foo""
        ""KijiTable xooTable = pool.get("xoo");""
        """"
        ""// read the table.""
        """"
        ""    KijiTableReader reader = fooTable1.openTableReader();""

        ""    KijiDataRequest dataRequest = KijiDataRequest.builder()""
        ""        .addColumns(ColumnsDef.create().add("some_family", "some_qualifier"))""
        ""        .build();""

        ""        // process one row""
        ""        KijiRowData rowData = reader.get(entityId, dataRequest);""
        """"
        ""        //process rows""
        ""        KijiScannerOptions scanOptions = new KijiScannerOptions()""
        ""            .setStartRow(table.getEntityId("the-start-row"))""
        ""            .setStopRow(table.getEntityId("the-stop-row"));""
        ""        final KijiRowScanner scanner = reader.getScanner(dataRequest, scanOptions);""
        ""          // Scan over the requested row range, in order:""
        ""        for (KijiRowData row : scanner) {""
            ""        }""
        ""        scanner.close();""
        ""    reader.close();""
        """"
        """"
        ""fooTable1.release();""
        ""fooTable2.release();""
        ""xooTable.release();""
        ""pool.close(); // force close all the unused(pooled) table connectins in the pool""
        ""kiji.release();""*/
    }

    public static void main(String[] args) {
        //Test.run(args);
        Test.test();
    }
}
